package org.gemini.transport.netty.channel;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.gemini.common.util.Constants;
import org.gemini.common.util.SystemClock;
import org.gemini.transport.Directory;
import org.gemini.transport.UnresolvedAddress;
import org.gemini.transport.channel.JChannelGroup;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static org.gemini.common.util.Preconditions.checkNotNull;

/**
 * 同一个地址下的channel 集合
 * next() 和 add() 方法是典型的生产者消费者模型
 * 在多线程情况下是需要考虑 next() 和 add()的线程同步问题，而且需要考虑防止假唤醒的情况（通过自旋锁控制）
 * Gemini
 * org.gemini.transport.netty.channel.NettyChannelGroup
 *
 * @author hailin.zhl
 * @date 2018/09/07
 */
public class NettyChannelGroup implements JChannelGroup {

    private static long LOSS_INTERVAL = TimeUnit.MINUTES.toMillis(5);

    private static int DEFAULT_SEQUENCE_STEP = (Constants.AVAILABLE_PROCESSORS << 3) + 1;

    private static final AtomicIntegerFieldUpdater<NettyChannelGroup> signalNeededUpdater =
            AtomicIntegerFieldUpdater.newUpdater(NettyChannelGroup.class, "signalNeeded");

    private final ConcurrentLinkedQueue<Runnable> waitAvailableListeners = new ConcurrentLinkedQueue<>();

    private final UnresolvedAddress address;

    private final CopyOnWriteArrayList<Channel> channels = new CopyOnWriteArrayList<>();

    // 连接断开时自动被移除
    private final ChannelFutureListener remover = new ChannelFutureListener() {

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            remove(future.channel());
        }
    };

    //private final IntSequence sequence = new IntSequence(DEFAULT_SEQUENCE_STEP);
    private AtomicInteger sequence = new AtomicInteger(0);

    private final ConcurrentMap<String, Integer> weights = new ConcurrentHashMap<>();

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notifyCondition = lock.newCondition();
    // attempts to elide conditional wake-ups when the lock is uncontended.
    @SuppressWarnings("all")
    private volatile int signalNeeded = 0; // 0: false, 1: true

    private volatile boolean connecting = false;

    private volatile int capacity = Integer.MAX_VALUE;
    private volatile int warmUp = Constants.DEFAULT_WARM_UP; // warm-up time
    private volatile long timestamp = SystemClock.millisClock().now();
    private volatile long deadlineMillis = -1;

    public NettyChannelGroup(UnresolvedAddress address) {
        this.address = address;
    }


    @Override
    public UnresolvedAddress remoteAddress() {
        return address;
    }

    @Override
    public Channel next() {
        for (; ; ) {
            int size = channels.size();
            if (size == 0) {
                if (waitForAvailable(1000)) { // wait a moment
                    continue;
                }
                throw new IllegalStateException("No channel");
            }
            if (size == 1) {
                return channels.get(0);
            }

            int index = sequence.getAndIncrement() & Integer.MAX_VALUE;
            return channels.get(index % size);
        }
    }

    @Override
    public List<? extends Channel> channels() {
        return new ArrayList<>(channels);
    }

    @Override
    public boolean isEmpty() {
        return channels.isEmpty();
    }

    @Override
    public boolean add(Channel channel) {
        boolean added = channels.add(channel);
        if (added) {
            timestamp = SystemClock.millisClock().now(); // reset timestamp

            channel.closeFuture().addListener(remover);
            deadlineMillis = -1;
            //Atomically sets the field of the given object managed by this updater to the given value and returns the old value.
            if (signalNeededUpdater.getAndSet(this, 0) != 0) { // signal needed: true
                final ReentrantLock _look = lock;
                _look.lock();
                try {
                    notifyCondition.signalAll(); // must signal all
                } finally {
                    _look.unlock();
                }
            }

            notifyListeners();
        }
        return added;
    }

    @Override
    public boolean remove(Channel channel) {
        boolean removed = channels.remove(channel);
        if (removed) {
            timestamp = SystemClock.millisClock().now(); // reset timestamp

            if (channels.isEmpty()) {
                deadlineMillis = SystemClock.millisClock().now() + LOSS_INTERVAL;
            }
        }
        return removed;
    }

    @Override
    public int size() {
        return channels.size();
    }

    @Override
    public void setCapacity(int capacity) {
        this.capacity = capacity;
    }

    @Override
    public int getCapacity() {
        return capacity;
    }

    @Override
    public boolean isConnecting() {
        return connecting;
    }

    @Override
    public void setConnecting(boolean connecting) {
        this.connecting = connecting;
    }

    @Override
    public boolean isAvailable() {
        return !channels.isEmpty();
    }

    @Override
    public boolean waitForAvailable(long timeoutMillis) {
        boolean available = isAvailable();
        if (available) {
            return true;
        }
        long remains = TimeUnit.MILLISECONDS.toNanos(timeoutMillis);

        final ReentrantLock _look = lock;
        _look.lock();
        try {
            // 通过自旋防止假唤醒
            // avoid "spurious wakeup" occurs
            while (!(available = isAvailable())) {
                signalNeeded = 1; // set signal needed to true
                if ((remains = notifyCondition.awaitNanos(remains)) <= 0) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            //TODO
        } finally {
            _look.unlock();
        }

        return available;
    }

    @Override
    public void onAvailable(Runnable listener) {
        waitAvailableListeners.add(listener);
        if (isAvailable()) {
            notifyListeners();
        }
    }

    @Override
    public int getWeight(Directory directory) {
        checkNotNull(directory, "directory");

        Integer weight = weights.get(directory.directoryString());
        return weight == null ? Constants.DEFAULT_WEIGHT : weight;
    }

    @Override
    public void putWeight(Directory directory, int weight) {
        checkNotNull(directory, "directory");

        if (weight == Constants.DEFAULT_WEIGHT) {
            // the default value does not need to be stored
            return;
        }
        weights.put(directory.directoryString(), weight > Constants.MAX_WEIGHT ? Constants.MAX_WEIGHT : weight);
    }

    @Override
    public void removeWeight(Directory directory) {
        checkNotNull(directory, "directory");

        weights.remove(directory.directoryString());
    }

    @Override
    public int getWarmUp() {
        return warmUp > 0 ? warmUp : 0;
    }

    @Override
    public void setWarmUp(int warmUp) {
        this.warmUp = warmUp;
    }

    @Override
    public boolean isWarmUpComplete() {
        return SystemClock.millisClock().now() - timestamp > warmUp;
    }

    @Override
    public long timestamp() {
        return timestamp;
    }

    @Override
    public long deadlineMillis() {
        return deadlineMillis;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        NettyChannelGroup that = (NettyChannelGroup) o;

        return address.equals(that.address);
    }

    @Override
    public int hashCode() {
        return address.hashCode();
    }

    @Override
    public String toString() {
        return "NettyChannelGroup{" +
                "address=" + address +
                ", channels=" + channels +
                ", weights=" + weights +
                ", warmUp=" + warmUp +
                ", timestamp=" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSZ").format(new Date(timestamp)) +
                ", deadlineMillis=" + deadlineMillis +
                '}';
    }


    void notifyListeners() {
        for (;;) {
            Runnable listener = waitAvailableListeners.poll();
            if (listener == null) {
                break;
            }
            listener.run();
        }
    }

    public static void main(String[] args) {
        NettyChannelGroup a = new NettyChannelGroup(new UnresolvedAddress("192.168.0.1", 80));
        signalNeededUpdater.compareAndSet(a, 0, 3);
        System.out.println(a.signalNeeded);
    }
}
